package com.ilikesoup.instant.split;

import com.google.common.collect.Lists;
import com.ilikesoup.instant.empty.EmptyUtils;
import org.slf4j.LoggerFactory;

import java.util.Collections;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Function;

/**
 * 分页任务
 *
 * @param <T>
 * @param <V>
 */
public class PageTask<T, V> implements Callable<List<V>> {
    /**
     * 存放当前页码
     */
    private final AtomicInteger pageNo;
    /**
     * 线程池
     */
    private final ThreadPoolExecutor executorService;
    /**
     * 获取指定页数据的函数，当返回列表为空时，视为后面再没有可用的数据
     */
    private final Function<Integer /* pageNo */, List<T>> pageListProducer;
    /**
     * 处理每页数据的函数
     */
    private final Function<List<T>, V> execution;
    /**
     * 线程数
     */
    private final int threads;
    /**
     * 分页任务是否已经结束，已有页数已全部获取
     */
    private volatile boolean isOver = false;
    /**
     * 一页一页地处理
     */
    private boolean pageByPage;

    public PageTask(Function<Integer /* pageNo */, List<T>> pageListProducer, Function<List<T>, V> execution) {
        this(pageListProducer, execution, 1);
    }

    public PageTask(Function<Integer /* pageNo */, List<T>> pageListProducer, Function<List<T>, V> execution, int startPage) {
        this.pageListProducer = pageListProducer;
        this.execution = execution;
        this.pageNo = new AtomicInteger(startPage);
        this.executorService = null;
        this.threads = 1;
    }

    public PageTask(ThreadPoolExecutor executorService, Function<Integer /* pageNo */, List<T>> pageListProducer, Function<List<T>, V> execution) {
        this(executorService, pageListProducer, execution, executorService.getMaximumPoolSize() - 1, 1);
    }

    public PageTask(ThreadPoolExecutor executorService, Function<Integer /* pageNo */, List<T>> pageListProducer, Function<List<T>, V> execution, int threads, int startPage) {
        this.pageNo = new AtomicInteger(startPage);
        this.executorService = executorService;
        this.pageListProducer = pageListProducer;
        this.execution = execution;
        if(threads <= 0) {
            threads = 2;
        }
        if(threads > executorService.getMaximumPoolSize()) {
            threads = executorService.getMaximumPoolSize();
        }
        this.threads = threads;
    }

    public List<V> call() throws InterruptedException {
        if(executorService == null || threads == 1) {
            // 单线程执行
            try {
                return new PageExecutor().call();
            } catch (Exception e) {
                e.printStackTrace();
            }
        } else if (pageByPage) {
            // 先取出一页数据，将数据分为若干等份，多线程并行执行，全部完成后再取下一页。
            List<T> pageList;
            List<V> result = Lists.newArrayList();
            while(EmptyUtils.isNotEmpty((pageList = fetchNextPage()))) {
                List<Future<V>> futureList = Lists.newArrayListWithCapacity(threads);
                int num = (pageList.size() + threads - 1) / threads;
                for (int i = 0; i < threads && i * num < pageList.size(); i++) {
                    List<T> each = Lists.newArrayList(pageList.subList(i * num, Math.max((i + 1) * num, pageList.size())));
                    futureList.add(executorService.submit(() -> execution.apply(each)));
                }
                for (Future<V> future : futureList) {
                    try {
                        result.add(future.get());
                    } catch (ExecutionException e) {

                    }
                }
            }
            return result;
        } else {
            // 并行执行，每个线程取一页数据，执行完后接着取，执行过程中不需要等待其他线程执行完成
            List<Future<List<V>>> futureList = Lists.newArrayListWithCapacity(threads);
            for (int i = 0; i < threads; i++) {
                futureList.add(executorService.submit(new PageExecutor()));
            }
            List<V> result = Lists.newArrayList();
            for (Future<List<V>> future : futureList) {
                try {
                    result.addAll(future.get());
                } catch (ExecutionException e) {
                    e.printStackTrace();
                }
            }
            return result;
        }
        return Collections.emptyList();
    }

    private class PageExecutor implements Callable<List<V>> {
        private List<V> executeResult = Lists.newArrayList();
        @Override
        public List<V> call() throws Exception {
            List<T> page;
            while(EmptyUtils.isNotEmpty((page = fetchNextPage()))) {
                executeResult.add(execution.apply(page));
            }
            return executeResult;
        }
    }

    private List<T> fetchNextPage() {
        if(isOver) {
            return Collections.emptyList();
        }
        List<T> page = pageListProducer.apply(pageNo.getAndIncrement());
        if(EmptyUtils.isEmpty(page)) {
            isOver = true;
        }
        return page;
    }

    public int getThreads() {
        return threads;
    }

}